package com.lkd.client.emq;

import com.alibaba.fastjson.JSON;
import com.lkd.client.config.EmqConfig;

import com.lkd.client.emq.msg.*;
import com.lkd.client.service.DataProcessService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * emq回调类
 */
@Component
@Slf4j
public class EmqMsgCallback implements MqttCallbackExtended {
    @Autowired
    private EmqClient emqClient;

    @Autowired
    private  EmqConfig config;

    @Autowired
    private  DataProcessService dataProcessService;

    @Override
    public void connectionLost(Throwable throwable) {
        log.info("emq connect lost",throwable);
        try {
            TimeUnit.SECONDS.sleep(10);
            //连接中断，从新建立连接
            emqClient.connect(this);
            //订阅服务器通知topic
            emqClient.subscribe(config.getSubscribeTopic(),this);
        } catch (Exception e) {
          log.info("订阅失败",e);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        CompletableFuture.runAsync(()->{
            String payload = new String(mqttMessage.getPayload());
            //出货通知
            if( topic.indexOf("/vendout")>=0 ){
                VendoutContract vendoutContract= JSON.parseObject(payload, VendoutContract.class);  //出货协议
                log.info("出货通知 vendoutReq {}",vendoutContract);
                dataProcessService.vendoutReq(vendoutContract);
            }
        });
    }


    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        CompletableFuture.runAsync(()->{
            try {
                MqttMessage mqttMessage= iMqttDeliveryToken.getMessage();
                String payload = new String(mqttMessage.getPayload());
                log.info("-------------deliveryComplete-------------{}",payload);
                VendoutResultContract vendoutResultContract = JSON.parseObject(payload, VendoutResultContract.class);
                if( vendoutResultContract.isSuccess() ){  //如果出货成功
                    dataProcessService.vendoutComplete(vendoutResultContract.getVendoutData().getOrderNo());  //完成上报
                }
            } catch (Exception e) {
                log.error("deliveryComplete",e);
            }
        });
    }

    @Override
    public void connectComplete(boolean connectComplete, String s) {
        //订阅服务器topic,及回调处理方法
        try {
            log.info("connectComplete subscribe {} ",config.getSubscribeTopic());
            emqClient.subscribe(config.getSubscribeTopic(),this);
        } catch (MqttException e) {
            log.error("connectComplete",e);
            e.printStackTrace();
        }
    }
}
